package com.atguigu.chapter06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/5 16:10
 */
public class Flink09_OrderTxDetect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 1.读取数据
        SingleOutputStreamOperator<OrderEvent> orderDS = env
                .readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new OrderEvent(
                                Long.valueOf(datas[0]),
                                datas[1],
                                datas[2],
                                Long.valueOf(datas[3])
                        );
                    }
                });


        SingleOutputStreamOperator<TxEvent> txDS = env
                .readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new TxEvent(
                                datas[0],
                                datas[1],
                                Long.valueOf(datas[2])
                        );
                    }
                });

        // TODO 实时对账： 关联两条流
//        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDS.connect(txDS);
        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDS.keyBy(order -> order.getTxId())
                .connect(txDS.keyBy(tx -> tx.getTxId()));

        // 通过 交易码进行 关联
        orderTxCS
//                .keyBy(order -> order.getTxId(), tx -> tx.getTxId())   // keyby方式一
                .process(new OrderTxDetectFunction())
                .print();

        env.execute();
    }

    public static class OrderTxDetectFunction extends CoProcessFunction<OrderEvent, TxEvent, String> {

        //存 业务系统 的数据 ： key是交易码，value是数据本身
        Map<String, OrderEvent> orderMap = new HashMap<>();
        //存 交易系统 的数据 ： key是交易码，value是数据本身
        Map<String, TxEvent> txMap = new HashMap<>();


        /**
         * 业务系统数据的处理函数
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            // 进入这个方法，来的数据是   业务系统的数据
            // 1.查： 查一下 交易系统的数据（相同交易码） 来过没有
            TxEvent txEvent = txMap.get(value.getTxId());
            if (txEvent != null) {
                // 2.1 表示 交易数据 来过
                // 对账成功
                out.collect("订单" + value.getOrderId() + "对账成功！");
                // 把之前保存的 交易数据 删除
                txMap.remove(value.getTxId());
            } else {
                // 2.2 表示 交易数据 没来过
                // 把 自己 存起来（业务数据）
                orderMap.put(value.getTxId(), value);
            }

        }

        /**
         * 交易系统数据的处理函数
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
            // 进入这个方法，来的数据是   交易系统的数据
            // 1.查： 查一下 业务系统的数据（相同交易码） 来过没有
            OrderEvent orderEvent = orderMap.get(value.getTxId());
            if (orderEvent != null) {
                // 2.1 表示 业务数据 来过
                // 对账成功
                out.collect("订单" + orderEvent.getOrderId() + "对账成功！");
                // 把之前保存的 业务数据 删除
                orderMap.remove(value.getTxId());
            } else {
                // 2.2 表示 业务数据 没来过
                // 把 自己 存起来（交易数据）
                txMap.put(value.getTxId(), value);
            }
        }
    }

}
/*
    如果并行度大于1，结果不太准确：因为 相同交易码的数据 ，可能在不同的分区，找不到对方
    => 解决： 对 交易码 进行keyby，这样 相同交易码的数据，就会进入同一个分区

    在开发中，使用connect，一般都会进行 keyby。
    先 keyby再 connect， 或者 connect之后再 keyby，都可以，效果一样
 */